package air

import (
	"net"
	"strings"
	"sync"
	"sync/atomic"

	"gitee.com/sdxstar/air/logger"
	"gitee.com/sdxstar/air/options"

	"github.com/pkg/errors"
)

type Server struct {
	opts      *options.ServerOptions
	listener  net.Listener
	exitChan  chan struct{}
	waitGroup sync.WaitGroup

	running uint32

	idGenerator *IdGenerator

	middleware []MiddlewareFunc

	routers     map[string]Router
	routersLock sync.RWMutex

	eventHandlers     map[Event]EventHandler
	eventHandlersLock sync.RWMutex

	msgHandlers     map[string]EventHandler
	msgHandlersLock sync.RWMutex

	conns     map[string]map[*connection]struct{}
	rooms     map[*connection]map[string]struct{}
	connsLock sync.RWMutex

	sids     map[string]*connection
	sidsLock sync.RWMutex

	log logger.Logger
}

func NewServer(opts *options.ServerOptions) (srv *Server, err error) {
	srv = &Server{
		opts:          opts,
		exitChan:      make(chan struct{}),
		eventHandlers: make(map[Event]EventHandler),
		msgHandlers:   make(map[string]EventHandler),
		conns:         make(map[string]map[*connection]struct{}),
		rooms:         make(map[*connection]map[string]struct{}),
		sids:          make(map[string]*connection),
		routers:       make(map[string]Router),
	}
	srv.buildOpts()

	srv.idGenerator, err = NewSnowFlake(srv.opts.WorkId)
	if err != nil {
		return nil, errors.Wrap(err, "init id generator failed")
	}

	srv.listener, err = net.Listen("tcp", srv.opts.Address)
	return
}

func (s *Server) buildOpts() {
	defaultOpt := options.DefaultServerOptions()
	if s.opts == nil {
		s.opts = defaultOpt
	}
	if s.opts.WorkId == 0 {
		s.opts.WorkId = defaultOpt.WorkId
	}
	if s.opts.Address == "" {
		s.opts.Address = defaultOpt.Address
	}
	if s.opts.HeartbeatTimeout == 0 {
		s.opts.HeartbeatTimeout = defaultOpt.HeartbeatTimeout
	}
	if s.opts.MessageTimeout == 0 {
		s.opts.MessageTimeout = defaultOpt.MessageTimeout
	}
	if s.opts.PacketSendChanLimit == 0 {
		s.opts.PacketSendChanLimit = defaultOpt.PacketSendChanLimit
	}
	if s.opts.PacketReceiveChanLimit == 0 {
		s.opts.PacketReceiveChanLimit = defaultOpt.PacketReceiveChanLimit
	}
	if s.opts.Codec == nil {
		s.opts.Codec = defaultOpt.Codec
	}
	if s.opts.Protocol == nil {
		s.opts.Protocol = defaultOpt.Protocol
	}
	if s.opts.Logger == nil {
		s.opts.Logger = logger.NewConsoleLog()
	}
	s.log = s.opts.Logger
}

// Start starts service
func (s *Server) Start() error {
	if s.Started() {
		return errors.WithStack(errors.New("server already started"))
	}
	s.log.Info("air server started on ", s.opts.Address)
	s.waitGroup.Add(1)
	defer s.waitGroup.Done()
	atomic.StoreUint32(&s.running, 1)
	for {
		select {
		case <-s.exitChan:
			return nil

		default:
		}

		conn, err := s.listener.Accept()
		if err != nil {
			if strings.Contains(err.Error(), ErrConnClosed.Error()) {
				return nil
			}
			return errors.Wrap(err, "air server accept failed")
		}

		s.waitGroup.Add(1)
		go func() {
			newConn(conn, s).start()
			s.waitGroup.Done()
		}()
	}
}

func (s *Server) SetLog(log logger.Logger) {
	s.log = log
}

func (s *Server) On(e string, fn EventHandler) {
	switch e {
	case OnConnect, OnHeartbeat, OnJoin, OnLeave, OnDisconnect, OnClose:
		s.eventHandlersLock.Lock()
		defer s.eventHandlersLock.Unlock()
		s.eventHandlers[e] = fn
	default:
		s.msgHandlersLock.Lock()
		defer s.msgHandlersLock.Unlock()
		s.msgHandlers[e] = fn
	}
}

func (s *Server) Use(middleware ...MiddlewareFunc) {
	s.middleware = append(s.middleware, middleware...)
}

func (s *Server) Group(m ...MiddlewareFunc) (g *Group) {
	g = NewGroup(s)
	g.Use(m...)
	return
}

func (e *Server) SetRouter(event string, router Router) {
	e.routersLock.Lock()
	defer e.routersLock.Unlock()
	e.routers[event] = router
}

func (e *Server) findRouter(event string) Router {
	if len(e.routers) > 0 {
		if r, ok := e.routers[event]; ok {
			return r
		}
	}
	return e
}

func (s *Server) Handler(e string) EventHandler {
	s.msgHandlersLock.Lock()
	defer s.msgHandlersLock.Unlock()
	return s.msgHandlers[e]
}

func (s *Server) middlewares() []MiddlewareFunc {
	return s.middleware
}

func (s *Server) BroadcastToAll(event string, msg []byte) {
	s.sidsLock.RLock()
	defer s.sidsLock.RUnlock()
	for _, conn := range s.sids {
		go conn.Emit(event, msg)
	}
}

func (s *Server) BroadcastTo(room, event string, msg []byte) {
	s.connsLock.RLock()
	defer s.connsLock.RUnlock()

	roomConns, ok := s.conns[room]
	if !ok {
		return
	}

	for conn := range roomConns {
		go conn.Emit(event, msg)
	}
}

func (s *Server) Started() bool {
	return atomic.LoadUint32(&s.running) == 1
}

func (s *Server) Close() {
	s.log.Info("server shutting down...")
	close(s.exitChan)
	_ = s.listener.Close()
	atomic.StoreUint32(&s.running, 0)
	s.waitGroup.Wait()
	s.log.Info("server closed")
}
